// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: secure discovery server and client
// feature: support auth token
// feature: provide query interfaces
// feature: develop dispatcher
// feature: Split history as an independent common module
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// feature: discovery support client and server
// feature: evbs support report hostname processname and version
// feature: add statictic lookup
// feature: dds support get remote participant vbs version and rm some print
// ------------------------------------------------------------------

#include <discovery/participant/PDPSimple.h>

#include <discovery/participant/PDPListener.h>
#include <discovery/participant/simple/SimplePDPEndpoints.hpp>
#include <discovery/participant/simple/SimplePDPEndpointsSecure.hpp>
#include <discovery/endpoint/EDPSimple.h>
#include <discovery/endpoint/EDPStatic.h>
#include <discovery/participant/DS/PDPSecurityInitiatorListener.hpp>
#include <edds/rtps/resources/TimedEvent.h>
#include <edds/rtps/builtin/BuiltinProtocols.h>
#include <edds/rtps/builtin/data/NetworkConfiguration.hpp>
#include <edds/rtps/builtin/liveliness/WLP.h>
#include <edds/rtps/builtin/data/ParticipantProxyData.h>
#include <edds/rtps/builtin/data/ReaderProxyData.h>
#include <edds/rtps/builtin/data/WriterProxyData.h>

#include <history/ParticipantListener.h>
#include <edds/rtps/writer/StatelessWriter.h>

#include <edds/rtps/reader/StatelessReader.h>
#include <edds/rtps/reader/StatefulReader.h>

#include <history/WriterHistory.h>
#include <history/ReaderHistory.h>

#include <ertps/utils/TimeConversion.h>
#include <ertps/utils/LocatorAlias.h>

#include <rtps/participant/RTPSParticipantImpl.h>

#include <mutex>

#include <edds/dds/builtin/lookup/LookupManager.hpp>
#include <history/TopicPayloadPoolRegistry.hpp>

#include <elog/Log.hpp>

using namespace evbs::ertps;

namespace evbs {
namespace ertps {
namespace rtps {

static HistoryAttributes pdp_reader_history_attributes(const BuiltinAttributes& builtin_att,
                                                       const RTPSParticipantAllocationAttributes& allocation) {
    HistoryAttributes hatt;
    hatt.payloadMaxSize = builtin_att.readerPayloadSize;
    hatt.memoryPolicy = builtin_att.readerHistoryMemoryPolicy;
    hatt.initialReservedCaches = 25;
    if (allocation.participants.initial > 0) {
        hatt.initialReservedCaches = (int32_t)allocation.participants.initial;
    }
    if (allocation.participants.maximum < std::numeric_limits<size_t>::max()) {
        hatt.maximumReservedCaches = (int32_t)allocation.participants.maximum;
    }

    return hatt;
}

static HistoryAttributes pdp_writer_history_attributes(const BuiltinAttributes& builtin_att) {
    HistoryAttributes hatt;
    hatt.payloadMaxSize = builtin_att.writerPayloadSize;
    hatt.memoryPolicy = builtin_att.writerHistoryMemoryPolicy;
    hatt.initialReservedCaches = 1;
    hatt.maximumReservedCaches = 1;

    return hatt;
}

PDPSimple::PDPSimple(BuiltinProtocols* built, const RTPSParticipantAllocationAttributes& allocation)
    : PDP(built, allocation) {}

PDPSimple::~PDPSimple() {}

void PDPSimple::update_builtin_locators() {
    auto endpoints = static_cast<ertps::rtps::SimplePDPEndpoints*>(builtin_endpoints_.get());
    mp_builtin->updateMetatrafficLocators(endpoints->reader.reader_->getAttributes().unicastLocatorList);
}

void PDPSimple::initializeParticipantProxyData(ParticipantProxyData* participant_data) {
    PDP::initializeParticipantProxyData(participant_data);

    const auto& discovery_config = getRTPSParticipant()->getAttributes().builtin.discovery_config;

    if (discovery_config.use_SIMPLE_EndpointDiscoveryProtocol) {
        if (discovery_config.m_simpleEDP.use_PublicationWriterANDSubscriptionReader) {
            participant_data->m_availableBuiltinEndpoints |= DISC_BUILTIN_ENDPOINT_PUBLICATION_ANNOUNCER;
            participant_data->m_availableBuiltinEndpoints |= DISC_BUILTIN_ENDPOINT_SUBSCRIPTION_DETECTOR;
        }

        if (discovery_config.m_simpleEDP.use_PublicationReaderANDSubscriptionWriter) {
            participant_data->m_availableBuiltinEndpoints |= DISC_BUILTIN_ENDPOINT_PUBLICATION_DETECTOR;
            participant_data->m_availableBuiltinEndpoints |= DISC_BUILTIN_ENDPOINT_SUBSCRIPTION_ANNOUNCER;
        }

#if HAVE_SECURITY
        if (discovery_config.m_simpleEDP.enable_builtin_secure_publications_writer_and_subscriptions_reader) {
            participant_data->m_availableBuiltinEndpoints |= DISC_BUILTIN_ENDPOINT_PUBLICATION_SECURE_ANNOUNCER;
            participant_data->m_availableBuiltinEndpoints |= DISC_BUILTIN_ENDPOINT_SUBSCRIPTION_SECURE_DETECTOR;
        }

        if (discovery_config.m_simpleEDP.enable_builtin_secure_subscriptions_writer_and_publications_reader) {
            participant_data->m_availableBuiltinEndpoints |= DISC_BUILTIN_ENDPOINT_SUBSCRIPTION_SECURE_ANNOUNCER;
            participant_data->m_availableBuiltinEndpoints |= DISC_BUILTIN_ENDPOINT_PUBLICATION_SECURE_DETECTOR;
        }
#endif  // if HAVE_SECURITY
    } else if (!discovery_config.use_STATIC_EndpointDiscoveryProtocol) {
        elogError(RTPS_PDP, RetCode_t::RETCODE_NOT_ENABLED,
                  "Neither EDP simple nor EDP static enabled. Endpoints will not be discovered.");
    }
}

bool PDPSimple::init(RTPSParticipantImpl* part) {
    // The DATA(p) must be processed after EDP endpoint creation
    if (!PDP::initPDP(part)) {
        return false;
    }
    // INIT EDP
    if (m_discovery.discovery_config.use_STATIC_EndpointDiscoveryProtocol) {
        try {
            mp_EDP = new EDPStatic(this, mp_RTPSParticipant);
        } catch (const foonathan::memory::out_of_memory& e) {
            (void)e;
            logWarning(RTPS_PDP, std::string("catch out_of_memory exception:") << e.what());
            return false;
        }
        if (!mp_EDP->initEDP(m_discovery)) {
            elogError(RTPS_PDP, RetCode_t::RETCODE_ERROR, "Endpoint discovery static configuration failed");
            delete mp_EDP;
            mp_EDP = nullptr;
            return false;
        }
    } else if (m_discovery.discovery_config.use_SIMPLE_EndpointDiscoveryProtocol) {
        try {
            mp_EDP = new EDPSimple(this, mp_RTPSParticipant);
        } catch (const foonathan::memory::out_of_memory& e) {
            (void)e;
            elogError(RTPS_PDP, RetCode_t::RETCODE_OUT_OF_RESOURCES,
                      std::string("catch out_of_memory exception:") << e.what());
            return false;
        }

        if (!mp_EDP->initEDP(m_discovery)) {
            elogError(RTPS_PDP, RetCode_t::RETCODE_ERROR, "Endpoint discovery simple configuration failed");
            delete mp_EDP;
            mp_EDP = nullptr;
            return false;
        }
    } else {
        logWarning(RTPS_PDP, "No EndpointDiscoveryProtocol defined");
        return false;
    }
    logDebug(RTPS_PDP, "PDP and EDP init successfully. Guid " << part->getGuid());
    return true;
}

ParticipantProxyData* PDPSimple::createParticipantProxyData(const ParticipantProxyData& participant_data,
                                                            const GUID_t&) {
    std::unique_lock<std::recursive_mutex> lock(*getMutex());

    // decide if we dismiss the participant using the ParticipantFilteringFlags
    const ParticipantFilteringFlags_t& flags = m_discovery.discovery_config.ignoreParticipantFlags;
    const GUID_t& remote = participant_data.m_guid;
    const GUID_t& local = getLocalParticipantProxyData()->m_guid;
    bool is_same_host =
        local.is_on_same_host_as(getLocalParticipantProxyData()->m_hostId, remote, participant_data.m_hostId);
    bool is_same_process =
        local.is_on_same_process_as(getLocalParticipantProxyData()->m_pid, remote, participant_data.m_pid);

    if (flags != ParticipantFilteringFlags_t::NO_FILTER) {
        if (!is_same_host) {
            if (flags & ParticipantFilteringFlags::FILTER_DIFFERENT_HOST) {
                return nullptr;
            }
        } else {
            bool filter_same = (flags & ParticipantFilteringFlags::FILTER_SAME_PROCESS) != 0;
            bool filter_different = (flags & ParticipantFilteringFlags::FILTER_DIFFERENT_PROCESS) != 0;

            if (filter_same && filter_different) {
                return nullptr;
            }

            if ((filter_same && is_same_process) || (filter_different && (!is_same_process))) {
                return nullptr;
            }
        }
    }

    ParticipantProxyData* pdata =
        add_participant_proxy_data(participant_data.m_guid, true, participant_data.m_startTime, &participant_data);
    if (pdata != nullptr) {
        (void)pdata->lease_duration_event->update_interval(pdata->m_leaseDuration);
        pdata->lease_duration_event->restart_timer();
        pdata->m_evbsRemoteVersion = participant_data.m_evbsRemoteVersion;
        pdata->m_hostName = participant_data.m_hostName;
    }

    return pdata;
}

// EDPStatic requires matching on ParticipantProxyData property updates
bool PDPSimple::updateInfoMatchesEDP() {
    return dynamic_cast<EDPStatic*>(mp_EDP) != nullptr;
}

void PDPSimple::announceParticipantState(bool new_change, bool dispose /* = false */) {
    WriteParams __wp = WriteParams::write_params_default();
    announceParticipantState(new_change, dispose, __wp);
}

void PDPSimple::announceParticipantState(bool new_change, bool dispose, WriteParams& wp) {
    if (enabled_) {
        new_change |= m_hasChangedLocalPDP.exchange(false);

#if HAVE_SECURITY
        if (mp_RTPSParticipant->is_secure()) {
            auto secure = dynamic_cast<ertps::rtps::SimplePDPEndpointsSecure*>(builtin_endpoints_.get());
            assert(nullptr != secure);

            WriterHistory& history = *(secure->secure_writer.history_);
            PDP::announceParticipantState(history, new_change, dispose, wp);
        }
#endif  // HAVE_SECURITY

        auto endpoints = dynamic_cast<ertps::rtps::SimplePDPEndpoints*>(builtin_endpoints_.get());
        WriterHistory& history = *(endpoints->writer.history_);
        PDP::announceParticipantState(history, new_change, dispose, wp);

        if (!dispose && !new_change && endpoints) {
            endpoints->writer.writer_->unsent_changes_reset();
        }
    }
}
bool PDPSimple::createPDPEndpoints() {
    ReaderListener* reader_listener = nullptr;
    logDebug(RTPS_PDP, "Beginning");

    ertps::rtps::SimplePDPEndpoints* endpoints = nullptr;
#if HAVE_SECURITY
    ertps::rtps::SimplePDPEndpointsSecure* secure_endpoints = nullptr;
    bool is_secure = mp_RTPSParticipant->is_secure();
    if (is_secure) {
        secure_endpoints = new ertps::rtps::SimplePDPEndpointsSecure();
        try {
            reader_listener = new PDPListener(this);
        } catch (const foonathan::memory::out_of_memory& e) {
            (void)e;
            elogError(RTPS_PDP, RetCode_t::RETCODE_OUT_OF_RESOURCES,
                      std::string("catch out_of_memory exception:") << e.what());
            delete secure_endpoints;
            secure_endpoints = nullptr;
            return false;
        }
        secure_endpoints->secure_reader.listener_.reset(reader_listener);

        endpoints = secure_endpoints;
        endpoints->reader.listener_.reset(
            new PDPSecurityInitiatorListener(this, [this](const ParticipantProxyData& participant_data) {
                match_pdp_remote_endpoints(participant_data, false, true);
            }));
    } else
#endif  // HAVE_SECURITY
    {
        endpoints = new ertps::rtps::SimplePDPEndpoints();
        try {
            reader_listener = new PDPListener(this);
        } catch (const foonathan::memory::out_of_memory& e) {
            (void)e;
            elogError(RTPS_PDP, RetCode_t::RETCODE_OUT_OF_RESOURCES,
                      std::string("catch out_of_memory exception:") << e.what());
            delete endpoints;
            endpoints = nullptr;
            return false;
        }
        endpoints->reader.listener_.reset(reader_listener);
    }
    builtin_endpoints_.reset(endpoints);

    bool ret_val = create_dcps_participant_endpoints();
#if HAVE_SECURITY
    if (ret_val && is_secure) {
        create_dcps_participant_secure_endpoints();
    }
#endif  // HAVE_SECURITY
    logDebug(RTPS_PDP, "SPDP Endpoints creation finished");
    return ret_val;
}

bool PDPSimple::create_dcps_participant_endpoints() {
    const RTPSParticipantAttributes& pattr = mp_RTPSParticipant->getAttributes();
    const RTPSParticipantAllocationAttributes& allocation = pattr.allocation;
    const BuiltinAttributes& builtin_att = mp_builtin->m_att;
    auto endpoints = dynamic_cast<ertps::rtps::SimplePDPEndpoints*>(builtin_endpoints_.get());
    assert(nullptr != endpoints);

    constexpr const char* topic_name = "DCPSParticipant";
    const EntityId_t reader_entity_id = c_EntityId_SPDPReader;
    const EntityId_t writer_entity_id = c_EntityId_SPDPWriter;

    // BUILTIN DCPSParticipant READER
    auto& reader = endpoints->reader;
    HistoryAttributes hatt;
    hatt = pdp_reader_history_attributes(builtin_att, allocation);

    PoolConfig reader_pool_cfg = PoolConfig::from_history_attributes(hatt);
    reader.payload_pool_ = TopicPayloadPoolRegistry::get(topic_name, reader_pool_cfg);
    reader.payload_pool_->reserve_history(reader_pool_cfg, true);
    reader.history_.reset(new ReaderHistory(hatt));

    ReaderAttributes ratt = create_builtin_reader_attributes();
    ratt.endpoint.reliabilityKind = BEST_EFFORT;

    RTPSReader* rtps_reader = nullptr;
    if (mp_RTPSParticipant->createReader(&rtps_reader, ratt, reader.payload_pool_, reader.history_.get(),
                                         reader.listener_.get(), reader_entity_id, true, false)) {
        reader.reader_ = dynamic_cast<StatelessReader*>(rtps_reader);
        assert(nullptr != reader.reader_);

#if HAVE_SECURITY
        mp_RTPSParticipant->set_endpoint_rtps_protection_supports(rtps_reader, false);
#endif  // if HAVE_SECURITY
        std::stringstream str;
        str << rtps_reader->getGuid().guidPrefix << "|" << rtps_reader->getGuid().entityId;
        rtps_reader->register_peroid_metrics(str.str());
    } else {
        elogError(RTPS_PDP, RetCode_t::RETCODE_ERROR, "'" << topic_name << "' builtin reader creation failed");
        reader.release();
        return false;
    }

    // BUILTIN DCPSParticipant WRITER
    auto& writer = endpoints->writer;
    hatt = pdp_writer_history_attributes(builtin_att);

    PoolConfig writer_pool_cfg = PoolConfig::from_history_attributes(hatt);
    writer.payload_pool_ = TopicPayloadPoolRegistry::get(topic_name, writer_pool_cfg);
    writer.payload_pool_->reserve_history(writer_pool_cfg, false);
    writer.history_.reset(new WriterHistory(hatt));

    WriterAttributes watt = create_builtin_writer_attributes();
    watt.endpoint.reliabilityKind = BEST_EFFORT;
    if (!m_discovery.initialPeersList.empty()) {
        // auto entry = LocatorSelectorEntry::create_fully_selected_entry(m_discovery.initialPeersList);
        // mp_RTPSParticipant->createSenderResources(entry);
        mp_RTPSParticipant->createSenderResources(m_discovery.initialPeersList);
    }

    RTPSWriter* rtps_writer = nullptr;
    if (mp_RTPSParticipant->createWriter(&rtps_writer, watt, writer.history_.get(), nullptr, writer_entity_id, true)) {
        writer.writer_ = dynamic_cast<StatelessWriter*>(rtps_writer);
        assert(nullptr != writer.writer_);
        std::stringstream str;
        str << rtps_writer->getGuid().guidPrefix << "|" << rtps_writer->getGuid().entityId;
        rtps_writer->register_peroid_metrics(str.str());
#if HAVE_SECURITY
        mp_RTPSParticipant->set_endpoint_rtps_protection_supports(rtps_writer, false);
#endif  // if HAVE_SECURITY

        vbs::transport::TransportFactory* network = mp_RTPSParticipant->network_factory();
        LocatorList_t fixed_locators;
        for (const Locator_t& loc : mp_builtin->m_initialPeersList) {
            if (!network->is_local_locator(mp_RTPSParticipant->get_transport_name(), loc) ||
                network->is_locator_allowed(mp_RTPSParticipant->get_transport_name(), loc)) {
                // Add initial peers locator without transformation as we don't know whether the
                // remote transport will allow localhost
                fixed_locators.push_back(loc);
            }
        }
        writer.writer_->set_fixed_locators(fixed_locators);
    } else {
        elogError(RTPS_PDP, RetCode_t::RETCODE_ERROR, "'" << topic_name << "' builtin writer creation failed");
        writer.release();
        return false;
    }
    return true;
}

#if HAVE_SECURITY
bool PDPSimple::create_dcps_participant_secure_endpoints() {
    const RTPSParticipantAttributes& pattr = mp_RTPSParticipant->getAttributes();
    const RTPSParticipantAllocationAttributes& allocation = pattr.allocation;
    const BuiltinAttributes& builtin_att = mp_builtin->m_att;
    auto endpoints = dynamic_cast<ertps::rtps::SimplePDPEndpointsSecure*>(builtin_endpoints_.get());
    assert(nullptr != endpoints);

    constexpr const char* topic_name = "DCPSParticipantsSecure";
    const EntityId_t reader_entity_id = c_EntityId_spdp_reliable_participant_secure_reader;
    const EntityId_t writer_entity_id = c_EntityId_spdp_reliable_participant_secure_writer;

    // BUILTIN DCPSParticipantsSecure READER
    auto& reader = endpoints->secure_reader;
    HistoryAttributes hatt;
    hatt = pdp_reader_history_attributes(builtin_att, allocation);

    PoolConfig reader_pool_cfg = PoolConfig::from_history_attributes(hatt);
    reader.payload_pool_ = TopicPayloadPoolRegistry::get(topic_name, reader_pool_cfg);
    reader.payload_pool_->reserve_history(reader_pool_cfg, true);
    reader.history_.reset(new ReaderHistory(hatt));

    ReaderAttributes ratt = create_builtin_reader_attributes();
    WriterAttributes watt = create_builtin_writer_attributes();
    add_builtin_security_attributes(ratt, watt);

    RTPSReader* rtps_reader = nullptr;
    if (mp_RTPSParticipant->createReader(&rtps_reader, ratt, reader.payload_pool_, reader.history_.get(),
                                         reader.listener_.get(), reader_entity_id, true, false)) {
        reader.reader_ = dynamic_cast<StatefulReader*>(rtps_reader);
        std::stringstream str;
        str << rtps_reader->getGuid().guidPrefix << "|" << rtps_reader->getGuid().entityId;
        rtps_reader->register_peroid_metrics(str.str());
        assert(nullptr != reader.reader_);
    } else {
        elogError(RTPS_PDP, RetCode_t::RETCODE_ERROR, "'" << topic_name << "' builtin reader creation failed");
        reader.release();
        return false;
    }

    // SPDP BUILTIN RTPSParticipant WRITER
    auto& writer = endpoints->secure_writer;
    hatt = pdp_writer_history_attributes(builtin_att);

    PoolConfig writer_pool_cfg = PoolConfig::from_history_attributes(hatt);
    writer.payload_pool_ = TopicPayloadPoolRegistry::get(topic_name, writer_pool_cfg);
    writer.payload_pool_->reserve_history(writer_pool_cfg, false);
    writer.history_.reset(new WriterHistory(hatt));

    RTPSWriter* rtps_writer = nullptr;
    if (mp_RTPSParticipant->createWriter(&rtps_writer, watt, writer.history_.get(), nullptr, writer_entity_id, true)) {
        writer.writer_ = dynamic_cast<StatefulWriter*>(rtps_writer);
        std::stringstream str;
        str << rtps_writer->getGuid().guidPrefix << "|" << rtps_writer->getGuid().entityId;
        rtps_writer->register_peroid_metrics(str.str());
        assert(nullptr != writer.writer_);
    } else {
        elogError(RTPS_PDP, RetCode_t::RETCODE_ERROR, "'" << topic_name << "' builtin writer creation failed");
        writer.release();
        return false;
    }
    return true;
}

#endif  // HAVE_SECURITY

void PDPSimple::assignRemoteEndpoints(ParticipantProxyData* pdata) {
    logDebug(RTPS_PDP, "For RTPSParticipant: " << pdata->m_guid.guidPrefix);
    bool ignored = false;
    notify_and_maybe_ignore_new_participant(pdata, ignored);
    if (!ignored) {
#if HAVE_SECURITY
        auto endpoints = dynamic_cast<ertps::rtps::SimplePDPEndpointsSecure*>(builtin_endpoints_.get());
        if (nullptr != endpoints) {
            // This participant is secure.
            // PDP should have been matched inside notifyAboveRemoteEndpoints after completing the authentication
            // process. We now match the other builtin endpoints.
            GUID_t remote_guid = pdata->m_guid;
            remote_guid.entityId = c_EntityId_spdp_reliable_participant_secure_writer;
            bool notify_secure = endpoints->secure_reader.reader_->matched_writer_is_matched(remote_guid);
            assign_low_level_remote_endpoints(*pdata, notify_secure);
        } else
#endif  // if HAVE_SECURITY
        {
            // This participant is not secure.
            // Match PDP and other builtin endpoints.
            match_pdp_remote_endpoints(*pdata, false, false);
            assign_low_level_remote_endpoints(*pdata, false);
        }
    }
}

void PDPSimple::removeRemoteEndpoints(ParticipantProxyData* pdata) {
    logDebug(RTPS_PDP, "For RTPSParticipant: " << pdata->m_guid);
    GUID_t guid = pdata->m_guid;

    {
        auto endpoints = static_cast<ertps::rtps::SimplePDPEndpoints*>(builtin_endpoints_.get());
        assert(nullptr != endpoints);

        guid.entityId = c_EntityId_SPDPWriter;
        endpoints->reader.reader_->matched_writer_remove(guid);

        guid.entityId = c_EntityId_SPDPReader;
        endpoints->writer.writer_->matched_reader_remove(guid);
    }

#if HAVE_SECURITY
    auto endpoints = dynamic_cast<ertps::rtps::SimplePDPEndpointsSecure*>(builtin_endpoints_.get());
    if (nullptr != endpoints) {
        guid.entityId = c_EntityId_spdp_reliable_participant_secure_writer;
        endpoints->secure_reader.reader_->matched_writer_remove(guid);

        guid.entityId = c_EntityId_spdp_reliable_participant_secure_reader;
        endpoints->secure_writer.writer_->matched_reader_remove(guid);
    }
#endif  // HAVE_SECURITY
}

void PDPSimple::notifyAboveRemoteEndpoints(const ParticipantProxyData& pdata, bool notify_secure_endpoints) {
    if (notify_secure_endpoints) {
        ParticipantProxyData pd = pdata;
        removeRemoteEndpoints(&pd);
        match_pdp_remote_endpoints(pdata, true, false);
    } else {
        // Add remote participant data
        GUID_t writer_guid {pdata.m_guid.guidPrefix, c_EntityId_SPDPWriter};
        ParticipantProxyData* part_data = createParticipantProxyData(pdata, writer_guid);
        if (part_data != nullptr) {
            bool ignored = false;
            notify_and_maybe_ignore_new_participant(part_data, ignored);
            if (!ignored) {
                match_pdp_remote_endpoints(*part_data, false, false);
                assign_low_level_remote_endpoints(*part_data, false);
            }
        }
    }
}

void PDPSimple::match_pdp_remote_endpoints(const ParticipantProxyData& pdata, bool notify_secure_endpoints,
                                           bool writer_only) {
#if !HAVE_SECURITY
    static_cast<void>(notify_secure_endpoints);
#endif  // !HAVE_SECURITY

    auto endpoints = static_cast<ertps::rtps::SimplePDPEndpoints*>(builtin_endpoints_.get());

    bool use_multicast_locators = !mp_RTPSParticipant->getAttributes().builtin.avoid_builtin_multicast ||
                                  pdata.metatraffic_locators.unicast.empty();
    const uint32_t endp = pdata.m_availableBuiltinEndpoints;

    // Default to values for non-secure endpoints
    auto reliability_kind = BEST_EFFORT_RELIABILITY_QOS;
    uint32_t pdp_reader_mask = ertps::rtps::DISC_BUILTIN_ENDPOINT_PARTICIPANT_DETECTOR;
    uint32_t pdp_writer_mask = ertps::rtps::DISC_BUILTIN_ENDPOINT_PARTICIPANT_ANNOUNCER;
    EntityId_t reader_entity_id = c_EntityId_SPDPReader;
    EntityId_t writer_entity_id = c_EntityId_SPDPWriter;

    RTPSReader* reader = endpoints->reader.reader_;
    RTPSWriter* writer = endpoints->writer.writer_;
#if HAVE_SECURITY
    // If the other participant has been authenticated, use values for secure endpoints
    if (notify_secure_endpoints) {
        auto secure_endpoints = static_cast<ertps::rtps::SimplePDPEndpointsSecure*>(builtin_endpoints_.get());
        reliability_kind = RELIABLE_RELIABILITY_QOS;
        pdp_reader_mask = ertps::rtps::DISC_BUILTIN_ENDPOINT_PARTICIPANT_SECURE_DETECTOR;
        pdp_writer_mask = ertps::rtps::DISC_BUILTIN_ENDPOINT_PARTICIPANT_SECURE_ANNOUNCER;
        reader_entity_id = c_EntityId_spdp_reliable_participant_secure_reader;
        writer_entity_id = c_EntityId_spdp_reliable_participant_secure_writer;
        reader = secure_endpoints->secure_reader.reader_;
        writer = secure_endpoints->secure_writer.writer_;
    }
#endif  // HAVE_SECURITY

    if (!writer_only && (0 != (endp & pdp_writer_mask))) {
        auto temp_writer_data = get_temporary_writer_proxies_pool().get();

        temp_writer_data->clear();
        temp_writer_data->guid().guidPrefix = pdata.m_guid.guidPrefix;
        temp_writer_data->guid().entityId = writer_entity_id;
        temp_writer_data->persistence_guid(pdata.get_persistence_guid());
        temp_writer_data->set_persistence_entity_id(writer_entity_id);
        temp_writer_data->set_remote_locators(pdata.metatraffic_locators, mp_RTPSParticipant, use_multicast_locators);
        temp_writer_data->m_qos.m_reliability.kind = reliability_kind;
        temp_writer_data->m_qos.m_durability.kind = TRANSIENT_LOCAL_DURABILITY_QOS;
#if HAVE_SECURITY
        if (notify_secure_endpoints) {
            if (!mp_RTPSParticipant->security_manager().discovered_builtin_writer(
                    reader->getGuid(), pdata.m_guid, *temp_writer_data,
                    reader->getAttributes().security_attributes())) {
                elogError(RTPS_EDP, RetCode_t::RETCODE_ERROR,
                          "Security manager returns an error for writer " << temp_writer_data->guid());
            }
        } else
#endif  // HAVE_SECURITY
        {
            reader->matched_writer_add(*temp_writer_data);
        }
    }

    if (0 != (endp & pdp_reader_mask)) {
        auto temp_reader_data = get_temporary_reader_proxies_pool().get();

        temp_reader_data->clear();
        // temp_reader_data->m_expectsInlineQos = false;
        temp_reader_data->guid().guidPrefix = pdata.m_guid.guidPrefix;
        temp_reader_data->guid().entityId = reader_entity_id;
        temp_reader_data->set_remote_locators(pdata.metatraffic_locators, mp_RTPSParticipant, use_multicast_locators);
        temp_reader_data->m_qos.m_reliability.kind = reliability_kind;
        temp_reader_data->m_qos.m_durability.kind = TRANSIENT_LOCAL_DURABILITY_QOS;
#if HAVE_SECURITY
        if (notify_secure_endpoints) {
            if (!mp_RTPSParticipant->security_manager().discovered_builtin_reader(
                    writer->getGuid(), pdata.m_guid, *temp_reader_data,
                    writer->getAttributes().security_attributes())) {
                elogError(RTPS_EDP, RetCode_t::RETCODE_ERROR,
                          "Security manager returns an error for reader " << temp_reader_data->guid());
            }
        } else
#endif  // HAVE_SECURITY
        {
            writer->matched_reader_add(*temp_reader_data);
        }

        if (!writer_only && (BEST_EFFORT_RELIABILITY_QOS == reliability_kind)) {
            endpoints->writer.writer_->unsent_changes_reset();
        }
    }
}

void PDPSimple::assign_low_level_remote_endpoints(const ParticipantProxyData& pdata, bool notify_secure_endpoints) {
    if (mp_EDP != nullptr) {
        mp_EDP->assignRemoteEndpoints(pdata, notify_secure_endpoints);
    }

    if (mp_builtin->mp_WLP != nullptr) {
        mp_builtin->mp_WLP->assignRemoteEndpoints(pdata, notify_secure_endpoints);
    }

    if (mp_builtin->tlm_ != nullptr) {
        mp_builtin->tlm_->assign_remote_endpoints(pdata);
    }
}

#if HAVE_SECURITY
bool PDPSimple::pairing_remote_writer_with_local_reader_after_security(const GUID_t& local_reader,
                                                                       const WriterProxyData& remote_writer_data) {
    auto endpoints = dynamic_cast<ertps::rtps::SimplePDPEndpointsSecure*>(builtin_endpoints_.get());
    if ((nullptr != endpoints) && (local_reader == endpoints->secure_reader.reader_->getGuid())) {
        endpoints->secure_reader.reader_->matched_writer_add(const_cast<WriterProxyData&>(remote_writer_data));
        return true;
    }

    return PDP::pairing_remote_writer_with_local_reader_after_security(local_reader, remote_writer_data);
}

bool PDPSimple::pairing_remote_reader_with_local_writer_after_security(const GUID_t& local_writer,
                                                                       const ReaderProxyData& remote_reader_data) {
    auto endpoints = dynamic_cast<ertps::rtps::SimplePDPEndpointsSecure*>(builtin_endpoints_.get());
    if ((nullptr != endpoints) && (local_writer == endpoints->secure_writer.writer_->getGuid())) {
        endpoints->secure_writer.writer_->matched_reader_add(const_cast<ReaderProxyData&>(remote_reader_data));
        return true;
    }

    return PDP::pairing_remote_reader_with_local_writer_after_security(local_writer, remote_reader_data);
}

#endif  // HAVE_SECURITY

bool PDPSimple::newRemoteEndpointStaticallyDiscovered(const GUID_t& pguid, int16_t userDefinedId, EndpointKind_t kind) {
    string_255 pname;
    if (lookup_participant_name(pguid, pname)) {
        if ((kind == WRITER) && (mp_EDP != nullptr)) {
            EDPStatic* EDPStatic_ptr = dynamic_cast<EDPStatic*>(mp_EDP);
            if (EDPStatic_ptr != nullptr) {
                (void)EDPStatic_ptr->newRemoteWriter(pguid, pname, static_cast<uint16_t>(userDefinedId));
            }
        } else if (mp_EDP != nullptr) {
            EDPStatic* EDPStatic_ptr = dynamic_cast<EDPStatic*>(mp_EDP);
            if (EDPStatic_ptr != nullptr) {
                (void)EDPStatic_ptr->newRemoteReader(pguid, pname, static_cast<uint16_t>(userDefinedId));
            }
        }
    }
    return false;
}

// Get all the participants in DataBase
bool PDPSimple::get_participants_in_db(std::vector<GuidPrefix_t>& prefix_list) {
    (void)prefix_list;
    elogWarning(RTPS_PDP, "The entity is Simple.");
    return false;
}

// Get the participant info by guid_prefix
bool PDPSimple::get_participant_info(const GuidPrefix_t& prefix, DiscoveryParticipantInfo& part_info) {
    (void)prefix;
    (void)part_info;
    elogWarning(RTPS_PDP, "The entity is Simple.");
    return false;
}

// Get the reader info by guid
bool PDPSimple::get_reader_info(const GUID_t& reader_guid, DiscoveryEndpointInfo& reader_info) {
    (void)reader_guid;
    (void)reader_info;
    elogWarning(RTPS_PDP, "The entity is Simple.");
    return false;
}

// Get the writer info by guid
bool PDPSimple::get_writer_info(const GUID_t& writer_guid, DiscoveryEndpointInfo& writer_info) {
    (void)writer_guid;
    (void)writer_info;
    elogWarning(RTPS_PDP, "The entity is Simple.");
    return false;
}

bool PDPSimple::get_topic_names(std::vector<std::string>& topic_list) {
    (void)topic_list;
    elogWarning(RTPS_PDP, "The entity is Simple.");
    return false;
}

bool PDPSimple::has_writer(const std::string& topic_name) {
    (void)topic_name;
    elogWarning(RTPS_PDP, "The entity is Simple.");
    return false;
}

bool PDPSimple::has_reader(const std::string& topic_name) {
    (void)topic_name;
    elogWarning(RTPS_PDP, "The entity is Simple.");
    return false;
}

bool PDPSimple::get_writer_of_topic(const std::string& topic_name,
                                    std::vector<DiscoveryEndpointInfo>& writer_info_list) {
    (void)topic_name;
    (void)writer_info_list;
    elogWarning(RTPS_PDP, "The entity is Simple.");
    return false;
}

bool PDPSimple::get_reader_of_topic(const std::string& topic_name,
                                    std::vector<DiscoveryEndpointInfo>& reader_info_list) {
    (void)topic_name;
    (void)reader_info_list;
    elogWarning(RTPS_PDP, "The entity is Simple.");
    return false;
}

bool PDPSimple::has_participant(const std::string& participant_name) {
    (void)participant_name;
    elogWarning(RTPS_PDP, "The entity is Simple.");
    return false;
}

bool PDPSimple::get_participant_of_name(const std::string& participant_name, DiscoveryParticipantInfo& part_info) {
    (void)participant_name;
    (void)part_info;
    elogWarning(RTPS_PDP, "The entity is Simple.");
    return false;
}
} /* namespace rtps */
} /* namespace ertps */
} /* namespace evbs */
